<?php
/**
* base.php
*
* Create on 2016-12-7
* Create by TT
*/
namespace haibao\cms\kafka;
class Base{
	
	const KAFKA_RESP_ERR_NO_ERROR = 0;
	const KAFKA_RESP_ERR__PARTITION_EOF = -191;
	const KAFKA_RESP_ERR__TIMED_OUT = -185;
	protected static $kafka = null;
	
	public function product($topicName = 'test', $message = ''){

		$this->initProducer();
		self::$kafka->product($topicName, $message);
	}
	
	public function consume($topicName = array('test','test1'), $consumeGroup = 'myConsumerGroup'){

		$this->initConsumer($topicName, $consumeGroup);
		return self::$kafka->consume();
	}
	

	protected function initProducer(){

		if(!self::$kafka){
			self::$kafka = $this->getKafka();
			self::$kafka->getProduct();
		}
		return self::$kafka->producer;
	}
	
	protected function initConsumer($topicName, $consumeGroup){

		if(!self::$kafka){
			self::$kafka = $this->getKafka();
			self::$kafka->getConsumer($topicName, $consumeGroup);
		}
		return self::$kafka->consumer;
	}
	
	private function getKafka(){
		$config = \haibao\classlibrary\www\Config::getConfig(\haibao\classlibrary\Config::KAFKA_HOST_LIST);
		if (!self::$kafka){
			self::$kafka  = new \haibao\frame\data\KafkaHelper();
			self::$kafka->setConfig($config);
		}
		return self::$kafka;
	}

}